package com.zhike.service.messages.impl;

import java.util.ArrayList;
import java.util.List;
import java.util.Vector;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import javax.annotation.Resource;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.BulkOperationException;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Repository;

import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.mongodb.MongoBulkWriteException;

import com.mongodb.client.result.UpdateResult;
import com.zhike.common.dto.DelAllMessageDTO;
import com.zhike.common.dto.MessagePageDTO;
import com.zhike.common.enums.YesNoEnum;
import com.zhike.common.repository.ClearMessage;
import com.zhike.common.repository.HistoryMessage;
import com.zhike.mapper.system.DeviceMapper;
import com.zhike.service.im.concurrent.CommonConcurrentExecutor;
import com.zhike.service.messages.EsSearchService;
import com.zhike.service.messages.IACKMessageRepository;
import com.zhike.service.messages.IClearMessageRepository;
import com.zhike.service.messages.IHistoryMessageRepository;

import lombok.extern.slf4j.Slf4j;

@Repository
@Slf4j
public class HistoryMessageRepository implements IHistoryMessageRepository {

    @Autowired
    private MongoTemplate mongoTemplate;

    @Resource
    private DeviceMapper deviceMapper;

    @Autowired
    private IClearMessageRepository clearMessageRepository;

    @Autowired
    private EsSearchService esSearchService;

    @Resource
    private IACKMessageRepository ackMessageRepository;

    private ScheduledExecutorService scheduledExecutorService;

    private static int BATCH_SIZE = 100;

    private List<HistoryMessage> dataBuffer = new Vector<>();

    // 每页最大数
    public static final int PAGE_LIMIT = 500;

    public static final int FLUSH_INTERVAL = 10; // 每30 秒刷新一次缓冲区

    public HistoryMessageRepository() {
        // 启动定时任务
        scheduledExecutorService = Executors.newScheduledThreadPool(1);
        scheduledExecutorService.scheduleAtFixedRate(this::insertBatch, FLUSH_INTERVAL, FLUSH_INTERVAL,
                TimeUnit.SECONDS);
        // 注册关闭钩子，当服务正常关闭时，确保缓冲区的数据被插入到数据库
        Runtime.getRuntime().addShutdownHook(new Thread(this::flush));
    }

    public void addHistoryMessage(HistoryMessage historyMessage) {
        mongoTemplate.insert(historyMessage);
//        dataBuffer.add(historyMessage);
//        if (dataBuffer.size() >= BATCH_SIZE) {
//            insertBatch();
//        }
    }

    private List<HistoryMessage> removeDuplicates(List<HistoryMessage> list) {
        return list.stream()
                .collect(Collectors.toMap(HistoryMessage::getFp, p -> p, (existing, replacement) -> existing))
                .values()
                .stream()
                .collect(Collectors.toList());
    }

    private void insertBatch() {
        if (!dataBuffer.isEmpty()) {
            try {
                dataBuffer = removeDuplicates(dataBuffer);
                mongoTemplate.insertAll(dataBuffer);
                CommonConcurrentExecutor.getInstance().execute(new Runnable() {
                    @Override
                    public void run() {
                        esSearchService.insertAll(dataBuffer);
                    }
                });
                ackMessageRepository.batchSaveAckMessageSingle(dataBuffer);
                dataBuffer.clear();
            } catch (Exception e) {
                if (e instanceof DuplicateKeyException) {
                    String message = e.getMessage();
                    if (null != message) {
                        log.debug("TEST1-1 buffer size:{}", dataBuffer.size());
                        int startIndex = message.indexOf("fp:");
                        int endIndex = message.indexOf("}");
                        if (startIndex != -1 && endIndex != -1) {
                            String duplicateValue = message.substring(startIndex + 3, endIndex);
                            duplicateValue = duplicateValue.replace("\"", "").trim();
                            for (int i = 0; i < dataBuffer.size(); i++) {
                                if (dataBuffer.get(i).getFp().equals(duplicateValue)) {
                                    dataBuffer.remove(i);
                                    log.debug("DEL1 {} size:{}", duplicateValue, dataBuffer.size());
                                }
                            }
                        } else {
                            log.debug("no found: {}", message);
                            int startIdx = message.indexOf("_id: \"");
                            int endIdx = message.indexOf("\" }'");
                            if (startIdx != -1 && endIdx != -1) {
                                String duplicateValue = message.substring(startIdx + 6, endIdx);
                                log.debug("find val: {} size:{}", duplicateValue, dataBuffer.size());
                                for (int i = 0; i < dataBuffer.size(); i++) {
                                    log.debug("{} {}", dataBuffer.get(i).getId(), dataBuffer.get(i));
                                    if (dataBuffer.get(i).getId().equals(duplicateValue)) {
                                        dataBuffer.remove(i);
                                        log.debug("DEL2 {}", duplicateValue);
                                    }
                                }
                            }
                        }
                        insertBatch();
                    }
                } else {
                    e.printStackTrace();
                    log.debug("TEST2 err: {}  err:{}", e.getClass(), e.getMessage());
                }
                return;
            }

        }
    }

    public void flush() {
        log.debug(">>>> 系统退出中,开始执行缓存数据存储");
        insertBatch();
    }

    @Override
    public void saveHistoryMessage(HistoryMessage historyMessage) {
        try {
            mongoTemplate.insert(historyMessage);
        } catch (Exception e) {
            log.error("save historyMessage err: {}", e.getMessage());
        }
    }

    /**
     * 返回一条满足条件的数据
     */
    @Override
    public HistoryMessage findHistoryMessageByFp(String fp) {
        Query query = new Query(Criteria.where("fp").is(fp)
                .and("deleted").is(YesNoEnum.NO.getValue())
                .and("revoked").is(YesNoEnum.NO.getValue()));
        return mongoTemplate.findOne(query, HistoryMessage.class);
    }

    @Override
    public List<HistoryMessage> findHistoryMessageBySession(String sessionId) {
        Query query = new Query(Criteria.where("sessionId").is(sessionId)
                .and("deleted").is(YesNoEnum.NO.getValue())
                .and("revoked").is(YesNoEnum.NO.getValue()));
        return mongoTemplate.find(query, HistoryMessage.class);
    }

    /**
     * 查询历史消息
     * 优先从缓存中查询，如果缓存中不存在，则从数据库中查询；
     * 默认每页返回最大200条数据；
     *
     * @param current 当前页码 0开始
     * @return
     */
    public List<HistoryMessage> findHistoryMessagePage(String uid, MessagePageDTO dto) {
        int size = 0; // 每页数量
        if (dto.getSize() == null) {
            size = PAGE_LIMIT;
        } else {
            size = dto.getSize();
        }

        // 排除已删除的记录
        List<ClearMessage> delRecord = clearMessageRepository.getDeleteRecord(dto.getSessionId(), uid, 1);
        List<String> delFps = new ArrayList<>();
        for (ClearMessage cm : delRecord) {
            delFps.add(cm.getFp());
        }

        List<HistoryMessage> list = new ArrayList<>(); // 缓存数据

        Criteria criteria = Criteria.where("sessionId").is(dto.getSessionId())
                .and("revoked").is(YesNoEnum.NO.getValue())
                .and("fp").nin(delFps);

        if (StringUtils.isNotBlank(dto.getFp())) {
            HistoryMessage hmLast = mongoTemplate.findOne(new Query(Criteria.where("fp").is(dto.getFp())),
                    HistoryMessage.class);
            if (hmLast != null) {
                if (dto.getAction().equals("up")) {
                    criteria.and("timestamp").lt(hmLast.getTimestamp());
                } else if (dto.getAction().equals("down")) {
                    criteria.and("timestamp").gt(hmLast.getTimestamp());
                } else if (dto.getAction().equals("pos")) {
                    Criteria c1 = Criteria.where("sessionId").is(dto.getSessionId())
                            .and("revoked").is(YesNoEnum.NO.getValue())
                            .and("fp").nin(delFps);
                    Query beforeQuery = new Query(c1.and("timestamp").lt(hmLast.getTimestamp()))
                            .with(Sort.by(Sort.Direction.DESC, "timestamp"))
                            .limit(5);
                    List<HistoryMessage> beforeRecords = mongoTemplate.find(beforeQuery, HistoryMessage.class);

                    Criteria c2 = Criteria.where("sessionId").is(dto.getSessionId())
                            .and("revoked").is(YesNoEnum.NO.getValue())
                            .and("fp").nin(delFps);
                    Query afterQuery = new Query(c2.and("timestamp").gt(hmLast.getTimestamp()))
                            .with(Sort.by(Sort.Direction.ASC, "timestamp"))
                            .limit(5);
                    List<HistoryMessage> afterRecords = mongoTemplate.find(afterQuery, HistoryMessage.class);

                    afterRecords.sort((u1, u2) -> u2.getTimestamp().compareTo(u1.getTimestamp()));
                    afterRecords.add(hmLast);
                    beforeRecords.addAll(0, afterRecords);
                    return beforeRecords;
                }
            }
        }

        // 大于清空消息的时间戳
        ClearMessage cm = clearMessageRepository.getClearRecord(dto.getSessionId(), uid);
        if (null != cm) {
            criteria.and("timestamp").gt(cm.getTimestamp());
        }

        Query query = new Query(criteria);
        query.limit(size);
        query.with(Sort.by(Sort.Order.desc("timestamp")));

        list = mongoTemplate.find(query, HistoryMessage.class);
        return list;
    }

    @Override
    @Deprecated
    public long updateReadStatus(String fromUid, String toUid) throws Exception {
        Query query = new Query(Criteria.where("senderUid").is(fromUid).and("receiver").is(toUid)
                .and("deleted").is(YesNoEnum.NO.getValue())
                .and("revoked").is(YesNoEnum.NO.getValue()));
        Update update = new Update();
        update.set("readStatus", YesNoEnum.YES.getValue());
        UpdateResult ur = mongoTemplate.updateMulti(query, update, HistoryMessage.class);
        return ur.getModifiedCount();
    }

    public IPage<HistoryMessage> adminfindHistoryMessagePage(String fakeId, Integer current, Integer pageSize) {
        if (current == null) {
            current = 0;
        }

        if (pageSize == null) {
            pageSize = PAGE_LIMIT;
        }

        Page<HistoryMessage> page = new Page<>(current, pageSize);

        long totalCount = 0; // 消息记录总数

        List<HistoryMessage> list = new ArrayList<>(); // 缓存数据

        if (list.size() == 0) {
            Pageable pageable = PageRequest.of((int) current, pageSize);

            Criteria criteria = new Criteria();

            criteria = Criteria.where("senderUid").is(fakeId);

            Query query = new Query(criteria);

            // 查询总数
            totalCount = mongoTemplate.count(query, HistoryMessage.class);

            // 增加分页条件
            query.with(pageable);
            query.with(Sort.by(Sort.Order.desc("timestamp")));

            list = mongoTemplate.find(query, HistoryMessage.class);
        }

        page.setTotal(totalCount).setRecords(list);
        return page;
    }

    @Override
    public boolean deleteAllMessage(DelAllMessageDTO dto) {

        List<HistoryMessage> hmList = this.findHistoryMessageBySession(dto.getSessionId());
        if (hmList == null || hmList.size() == 0) {
            return false;
        }
        int count = 0;
        for (HistoryMessage hm : hmList) {
            Update update = new Update();
            if (StringUtils.isBlank(hm.getDeleteUid())) {// 一方用户删除
                update.set("deleteUid", dto.getDeleteUid());
            } else if (hm.getDeleteUid().equals("0")) {
                continue;
            } else {
                // 避免同一人的重复清空
                if (dto.getDeleteUid().equals(hm.getDeleteUid())) {
                    continue;
                } else {
                    update.set("deleteUid", "0")
                            .set("deleted", YesNoEnum.YES.getValue());
                }
            }
            log.debug("待删除消息条数: {}", update.getUpdateObject().size());
            Query query = new Query(Criteria.where("_id").is(hm.getId()));
            UpdateResult ur = mongoTemplate.updateFirst(query, update, HistoryMessage.class);
            if (ur.getModifiedCount() > 0) {
                count++;
            }
        }

        if (count > 0) {
            log.debug("{} 清空了会话 {} 的历史消息 共 {} 条", dto.getDeleteUid(), dto.getSessionId(), count);
            return true;
        }

        return false;
    }

    @Override
    public Boolean revokeMessage(String fp) {
        Query query = new Query(Criteria.where("fp").is(fp)
                .and("deleted").is(YesNoEnum.NO.getValue())
                .and("revoked").is(YesNoEnum.NO.getValue()));

        Update update = new Update();
        update.set("revoked", YesNoEnum.YES.getValue());
        UpdateResult ur = mongoTemplate.updateMulti(query, update, HistoryMessage.class);
        return ur.getModifiedCount() > 0;
    }

    @Override
    public HistoryMessage findHistoryMessageByGroup(String groupId, String senderUid) {
        Query query = new Query(Criteria.where("groupId").is(groupId)
                .and("senderUid").is(senderUid));
        List<HistoryMessage> list = mongoTemplate.find(query, HistoryMessage.class);
        if (list != null) {
            return list.get(0);
        }
        return null;
    }

}
